package yredis

import (
	"context"
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/gotask/chantask"
	"github.com/go-redis/redis/v8"
	"sync"
	"sync/atomic"
)

type yredisCmdRef struct {
	refN  int32
	id    string
	redis *RedisInner
}

type yredisSubRef struct {
	workerN       int32
	id            string
	lk            sync.RWMutex
	topics        map[string]int8
	ex            *utils.Subscribe
	redis         *RedisInner
	redisSub      *redis.PubSub
	subClosedFlag int8
	subDoneFlag   int8
	pubQueId      string
}

type YredisGroup struct {
	cmdLk  sync.RWMutex
	cmdMap map[string]*yredisCmdRef

	subLk  sync.RWMutex
	subMap map[string]*yredisSubRef
}

func NewYredisGroup() *YredisGroup {
	rval := &YredisGroup{subMap: make(map[string]*yredisSubRef),
		cmdMap: make(map[string]*yredisCmdRef)}
	return rval
}

var (
	DefaultYredisGroup = NewYredisGroup()
)

func (this *YredisGroup) CheckGetRedisCmd(connstr string) interface{} {
	this.cmdLk.RLock()
	itm := this.cmdMap[connstr]
	if itm != nil {
		atomic.AddInt32(&itm.refN, 1)
	}
	this.cmdLk.RUnlock()
	if itm == nil {
		conf := gobase.NewStrMap()
		conf.URLFormDecode(connstr)
		addr := conf.StringByName("connstr", "")
		if len(addr) == 0 {
			return nil
		}
		user := conf.StringByName("user", "")
		pass := conf.StringByName("pass", "")

		this.cmdLk.Lock()
		itm = this.cmdMap[connstr]
		if itm == nil {
			itm = &yredisCmdRef{
				id:    connstr,
				redis: NewRedisInner(),
			}
			itm.redis.Config(addr, user, pass)
			this.cmdMap[connstr] = itm
		}
		atomic.AddInt32(&itm.refN, 1)
		this.cmdLk.Unlock()
	}
	return itm
}

func (this *YredisGroup) GetRedisInner(obj interface{}) *RedisInner {
	return obj.(*yredisCmdRef).redis
}

func (this *YredisGroup) Pub(obj interface{}, topic string, buf []byte) (int64, error) {
	itm := obj.(*yredisCmdRef)
	clt := itm.redis.CheckGetClt()
	if clt == nil {
		return 0, fmt.Errorf("连接参数不正确")
	}
	return clt.Publish(ctx_background, topic, buf).Result()
}

func (this *YredisGroup) ReleaseRedisCmd(obj interface{}) {
	itm := obj.(*yredisCmdRef)
	n := atomic.AddInt32(&itm.refN, -1)
	if n == 0 {
		delflag := 0
		this.cmdLk.Lock()
		if atomic.LoadInt32(&itm.refN) == 0 {
			delete(this.cmdMap, itm.id)
			delflag = 1
		}
		this.cmdLk.Unlock()
		if delflag == 1 {
			go itm.redis.Close()
		}
	}
}

// 根据connstr 获取一个订阅连接对象, 如果不存在则进行创建并初始化配置参数
// nil: 表示参数设定不完整, 无法进行创建
func (this *YredisGroup) checkGetRedisSub(connstr string) *yredisSubRef {
	this.subLk.RLock()
	itm := this.subMap[connstr]
	this.subLk.RUnlock()
	if itm == nil {
		conf := gobase.NewStrMap()
		conf.URLFormDecode(connstr)
		addr := conf.StringByName("connstr", "")
		if len(addr) == 0 {
			return nil
		}
		user := conf.StringByName("user", "")
		pass := conf.StringByName("pass", "")

		this.subLk.Lock()
		itm = this.subMap[connstr]
		if itm == nil {
			itm = &yredisSubRef{
				id:    connstr,
				redis: NewRedisInner(),
				ex:    utils.NewSubscribe(),
			}
			itm.pubQueId = conf.StringByName("pubque", "")
			itm.redis.Config(addr, user, pass)
			this.subMap[connstr] = itm
		}
		this.subLk.Unlock()
	}
	return itm
}

// 启动订阅
func (this *YredisGroup) checkStartSub(subref *yredisSubRef, topic string) error {
	sub := subref.redisSub
	if sub == nil { // 先确保创建成功
		subref.redis.CheckGetClt()
		sub = subref.redis.NewPSubscribe()
		subref.redisSub = sub
		ex := subref.ex

		go func() { // 移动接收线程, 无网络也可以创建成功
			if utils.GoFunCatchException {
				defer utils.PanicHandler()
			}
			atomic.AddInt32(&subref.workerN, 1)
			defer func() {
				atomic.AddInt32(&subref.workerN, -1)
				subref.subDoneFlag = 1
			}()
			var pubfn = func(topic string, buf []byte) {
				ex.Pub(topic, 0, buf)
			}
			taskid := subref.pubQueId
			if len(taskid) > 0 {
				pubQueChan, _ := chantask.DefaultChanTaskHub.CheckGetChanTask(taskid, 8192, 0)
				defer chantask.DefaultChanTaskHub.ReleaseChanTask(taskid)
				pubfn = func(topic string, buf []byte) {
					pubQueChan.PushTask(func(args ...interface{}) {
						topic1 := args[0].(string)
						buf1 := args[1].([]byte)
						ex.Pub(topic1, 0, buf1)
					}, topic, buf)
				}
			}
			for msg := range sub.Channel() {
				buf := []byte(msg.Payload)
				pubfn(msg.Channel, buf)
			}
		}()
	}

	//err := utils.DefaultWorkers().PostTaskFunc(subref.id, func() {
	//	// 无网络也可以订阅生效, 建立连接成功后, 也会生效
	//  sub.PSubscribe(context.Background(), topic)
	//})
	err := sub.PSubscribe(context.Background(), topic)
	return err
}

func (this *YredisGroup) GetSubCount() int {
	this.subLk.RLock()
	defer this.subLk.RUnlock()
	return len(this.subMap)
}

func (this *YredisGroup) GetStatusString() string {
	this.subLk.RLock()
	str1 := fmt.Sprintf("sub-n:%d", len(this.subMap))
	this.subLk.RUnlock()

	this.cmdLk.RLock()
	str2 := fmt.Sprintf("cmd-n:%d", len(this.cmdMap))
	this.cmdLk.RUnlock()

	return fmt.Sprintf("%s,%s", str1, str2)
}

func (this *YredisGroup) GetSubStatus(obj interface{}) string {
	itm := obj.(*yredisSubRef)
	if itm.subClosedFlag == 1 {
		return "closed"
	} else {
		return fmt.Sprintf("worker:%d, topics:%d, done:%d, closedflag:%d, msg:%s", itm.workerN, itm.ex.TopicCount(), itm.subDoneFlag, itm.subClosedFlag, itm.redis.StatusString())
	}
}

// 订阅通道数据
// 确保不会进行阻塞
// err=nil时确保会订阅成功
func (this *YredisGroup) Sub(connstr string, id string, topic string, fn utils.SubFunc) (obj interface{}, err error) {
	itm := this.checkGetRedisSub(connstr)
	if itm == nil {
		return nil, fmt.Errorf("创建Redis连接失败, 连接参数非法sample:connstr=ip:port&user=&pass&rand=0")
	}
	itm.lk.Lock()
	defer itm.lk.Unlock()
	if itm.subClosedFlag == 1 { // 通道关闭时, itm.lk会加写入锁
		return nil, fmt.Errorf("订阅失败, 通道正在关闭, 请稍后尝试")
	}

	// 内部对象进行订阅
	itm.ex.Sub(id, topic, fn)

	n := itm.ex.GetTopicSubCount(topic)
	if n == 1 { // 主题的第一次订阅
		err = this.checkStartSub(itm, topic)
		if err != nil {
			itm.ex.Unsub(id, topic)
		}
	}
	return itm, err
}

// 订阅通道数据
// 确保不会进行阻塞
// err=nil时确保会订阅成功
func (this *YredisGroup) SubEx(connstr string, id string, redisTopic, subtopic string, fn utils.SubFunc) (obj interface{}, err error) {
	itm := this.checkGetRedisSub(connstr)
	if itm == nil {
		return nil, fmt.Errorf("创建Redis连接失败, 连接参数非法sample:connstr=ip:port&user=&pass&rand=0")
	}
	itm.lk.Lock()
	defer itm.lk.Unlock()
	if itm.subClosedFlag == 1 { // 通道关闭时, itm.lk会加写入锁
		return nil, fmt.Errorf("订阅失败, 通道正在关闭, 请稍后尝试")
	}

	// 内部对象进行订阅
	itm.ex.Sub(id, subtopic, fn)

	n := itm.ex.GetTopicSubCount(subtopic)
	if n == 1 { // 主题的第一次订阅
		err = this.checkStartSub(itm, redisTopic)
		if err != nil {
			itm.ex.Unsub(id, subtopic)
		}
	}
	return itm, err
}

// 取消订阅通道数据
// 确保不会进行阻塞
func (this *YredisGroup) Unsub(obj interface{}, id string, topic string) (err error) {
	itm := obj.(*yredisSubRef)
	itm.lk.Lock()
	if itm.subClosedFlag == 0 {
		itm.ex.Unsub(id, topic)
		n := itm.ex.GetTopicSubCount(topic)
		if n == 0 {
			subref := itm
			sub := subref.redisSub
			go func() {
				if utils.GoFunCatchException {
					defer utils.PanicHandler()
				}
				sub.PUnsubscribe(context.Background(), topic)
			}()

			topicN := subref.ex.TopicCount()
			if topicN == 0 {
				subref.subClosedFlag = 1
				go func() {
					if utils.GoFunCatchException {
						defer utils.PanicHandler()
					}
					subref.redisSub.Close()
					subref.redis.Close()
				}()
			}
		}
	}
	itm.lk.Unlock()

	if itm.subClosedFlag == 1 { // 不在使用, 移除
		this.subLk.Lock()
		delete(this.subMap, itm.id)
		this.subLk.Unlock()
	}
	return
}
